﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Microsoft.ComplexEventProcessing;
using Microsoft.ComplexEventProcessing.Adapters;
using Microsoft.Samples.SmartGrid.AMI.MeterEventType;
using Microsoft.Samples.SqlServer;
using System.Threading;
using System.Data.SqlClient;
using System.IO;

namespace Microsoft.Samples.SmartGrid.AMI.ServiceBrokerOutputAdapter
{
    class ServiceBrokerOutputPointAdapter : PointOutputAdapter
    {
        private ServiceBrokerOutputAdapterConfig config;
        private CepEventType eventType;
        private SqlConnection conn = null;
        private Conversation dialog = null;
        private Service client = null;
        private SqlTransaction tran = null;

        public ServiceBrokerOutputPointAdapter(ServiceBrokerOutputAdapterConfig config, CepEventType type)
        {
            this.config = config;
            this.eventType = type;
        }

        public override void Resume()
        {
            new Thread(this.ConsumeEvents).Start();
        }

        public override void Start()
        {
            new Thread(this.ConsumeEvents).Start();
        }

        protected override void Dispose(bool disposing)
        {
            if (disposing)
            {
                this.conn.Close();
            }
            base.Dispose(disposing);
        }

        public void ConsumeEvents()
        {
            PointEvent currentEvent = default(PointEvent);
            conn = new SqlConnection(config.connString);
            conn.Open();
            
            while (true)
            {
                if (AdapterState.Stopping == AdapterState)
                {
                    Stopped();
                    return;
                }

                if (DequeueOperationResult.Empty == Dequeue(out currentEvent))
                {
                    Ready();
                    return;
                }

                if (currentEvent.EventKind == EventKind.Insert)
                {
                    string sbMessage = ConstructServiceBrokerMessage(currentEvent);
                    SendToServiceBrokerService(sbMessage);
                }

                ReleaseEvent(ref currentEvent);
            }
        }

        public string ConstructServiceBrokerMessage(PointEvent currentEvent)
        {
            string currentReading = currentEvent.GetField(0).ToString();
            string temp = currentEvent.GetField(1).ToString();
            string eventTimeStamp = currentEvent.GetField(2).ToString();
            string eventType = currentEvent.GetField(3).ToString();
            string meterHeadendID = currentEvent.GetField(4).ToString();
            string meterNumber = currentEvent.GetField(5).ToString();
            string meterVendorID = currentEvent.GetField(6).ToString();
           
            string sbMessage = "";
            sbMessage = sbMessage + "<MeterEvent>";
            sbMessage = sbMessage + "  <TimeStamp>" + eventTimeStamp.ToString() + "</TimeStamp>";
            sbMessage = sbMessage + "  <Type>" + eventType + "</Type>";
            sbMessage = sbMessage + "  <Data>";
            sbMessage = sbMessage + "    <CurrentReading>" + currentReading.ToString() + "</CurrentReading>";
            sbMessage = sbMessage + "    <Temp>" + temp.ToString() + "</Temp>";
            sbMessage = sbMessage + "  </Data>";
            sbMessage = sbMessage + "  <MeterInfo>";
            sbMessage = sbMessage + "    <Number>" + meterNumber + "</Number>";
            sbMessage = sbMessage + "    <VendorID>" + meterVendorID + "</VendorID>";
            sbMessage = sbMessage + "    <HeadendID>" + meterHeadendID + "</HeadendID>";
            sbMessage = sbMessage + "  </MeterInfo>";
            sbMessage = sbMessage + "</MeterEvent>";

            return sbMessage;
        }

        public void SendToServiceBrokerService(string sbMessage)
        {
            tran = conn.BeginTransaction();

            client = new Service(config.serviceName, conn, tran);
            client.FetchSize = 1;

            dialog = client.BeginDialog(
                config.serviceName, 
                null, 
                config.serviceContract, 
                TimeSpan.FromMinutes(1), 
                false, 
                conn, 
                tran);
            
            Message msg = new Message(config.messageType, new MemoryStream(Encoding.UTF8.GetBytes(sbMessage)));

            dialog.Send(msg,conn,tran);
            dialog.EndConversation(conn,tran);
            tran.Commit();
        }
    }
}
